package top.chenjipdc;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class SampleProducer {

    public static void main(String[] args) {

        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.251:9092");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "TestProducer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<>(properties);
        ProducerWrapper wrapper = new ProducerWrapper(producer, "mongo_logs", false);
        wrapper.start();
    }
}

class ProducerWrapper extends Thread {

    private final Producer<String, String> producer;
    private final String topic;
    private final Boolean isAsync;


    ProducerWrapper(Producer<String, String> kafkaProducer, String topic, Boolean isAsync) {
        this.producer = kafkaProducer;
        this.topic = topic;
        this.isAsync = isAsync;
    }

    @Override
    public void run(){
        int a = 0;
        while (true){
            String value = "" + a;
            String key = "" + (a * 2);
            if (isAsync){
                producer.send(new ProducerRecord<>(topic, key, value), (RecordMetadata metadata, Exception exception) -> {
                    if (exception != null){
                        throw new RuntimeException(exception);
                    }else {
                        System.out.println(metadata);
                    }
                });
            }else {
                try {
                    producer.send(new ProducerRecord<>(topic, key, value)).get();
                }catch (Exception e){
                    throw new RuntimeException(e);
                }
            }

            try {
                Thread.sleep(1000 * 5);
            }catch (Exception e){
                break;
            }
            a++;
            System.out.println("-------------------------------");
        }
        producer.close();

    }
}
